/*
 * FileName: MessageSplitter.java
 * Author:   zzw
 * Date:     2019年01月23日
 * Description:
 */
package com.chezhibao.flink.splitter;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple13;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 〈〉<br>
 * 〈〉
 *
 * @author zzw
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本]（可选）
 */
public class MessageSplitter implements FlatMapFunction<String, Tuple13<String, String,String,String,String,String,String,String,String,String,String,String,String>> {

    private Logger logger = LoggerFactory.getLogger(MessageSplitter.class);

    @Override
    public void flatMap(String s, Collector<Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>> collector) throws Exception {
        String[] datas = s.split("\\|");
        if(datas.length == 5){
            collector.collect(new Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>(datas[0], datas[1],datas[2], datas[3],datas[4], null, null, null, null, null,null, null, null));
        }
        else if(datas.length == 8){
            collector.collect(new Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>(datas[0], datas[1],datas[2], datas[3],datas[4], datas[5],datas[6], datas[7], null, null,null, null, null));
        }
        else if(datas.length == 9){
            collector.collect(new Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>(datas[0], datas[1],datas[2], datas[3],datas[4], datas[5],datas[6], datas[7],datas[8], null,null, null, null));
        }
        else if(datas.length == 12){
            collector.collect(new Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>(datas[0], datas[1],datas[2], datas[3],datas[4], datas[5],datas[6], datas[7],datas[8], datas[9],datas[10], datas[11], null));
        }
        else if(datas.length == 13){
            collector.collect(new Tuple13<String, String, String, String, String, String, String, String, String, String, String, String, String>(datas[0], datas[1],datas[2], datas[3],datas[4], datas[5],datas[6], datas[7],datas[8], datas[9],datas[10], datas[11], datas[12]));
        }
        else {
            logger.info("s:" + s);
        }
    }
}
